package com.gitee.ywj1352.reactor;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Random;

public class Learning01Create {

    //==========================================================================
    //----------------------静态生产逻辑------------------------------------------
    //==========================================================================
    @Test
    public void fun1() {
        Flux.interval(Duration.of(10, ChronoUnit.SECONDS)).subscribe(System.out::println);
    }

    /**
     * just()：可以指定序列中包含的全部元素。创建出来的 Flux 序列在发布这些元素之后会自动结束。
     * fromArray()，fromIterable()和 fromStream()：可以从一个数组、Iterable 对象或 Stream 对象中创建 Flux 对象。
     * empty()：创建一个不包含任何元素，只发布结束消息的序列。
     * error(Throwable error)：创建一个只包含错误消息的序列。
     * never()：创建一个不包含任何消息通知的序列。
     * range(int start, int count)：创建包含从 start 起始的 count 个数量的 Integer 对象的序列。
     * interval(Duration period)和 interval(Duration delay, Duration period)：创建一个包含了从 0 开始递增的 Long 对象的序列。其中包含的元素按照指定的间隔来发布。除了间隔时间之外，还可以指定起始元素发布之前的延迟时间。
     * intervalMillis(long period)和 intervalMillis(long delay, long period)：与 interval()方法的作用相同，只不过该方法通过毫秒数来指定时间间隔和延迟时间。
     */
    @Test
    public void staticCreateFun(){
        Flux.just("Hello", "World").subscribe(System.out::println);
        Flux.fromArray(new Integer[] {1, 2, 3}).subscribe(System.out::println);
        Flux.empty().subscribe(System.out::println);
        Flux.range(1, 10).subscribe(System.out::println);
        Flux.interval(Duration.of(10, ChronoUnit.SECONDS)).subscribe(System.out::println);
    }

    //==========================================================================
    //----------------------动态生产逻辑------------------------------------------
    //==========================================================================

    /**
     * generate()方法通过同步和逐一的方式来产生 Flux 序列。序列的产生是通过调用所提供的 SynchronousSink 对象的 next()，complete()和 error(Throwable)方法来完成的。
     * 逐一生成的含义是在具体的生成逻辑中，next()方法只能最多被调用一次。在有些情况下，序列的生成可能是有状态的，需要用到某些状态对象。此时可以使用 generate()方法的另外一种形式
     * generate(Callable<S> stateSupplier, BiFunction<S,SynchronousSink<T>,S> generator)，其中 stateSupplier 用来提供初始的状态对象。在进行序列生成时，状态对象会作为
     * generator 使用的第一个参数传入，可以在对应的逻辑中对该状态对象进行修改以供下一次生成时使用。
     *
     * 在代码清单 2中，第一个序列的生成逻辑中通过 next()方法产生一个简单的值，然后通过 complete()方法来结束该序列。如果不调用 complete()方法，所产生的是一个无限序列。第二个序列的
     * 生成逻辑中的状态对象是一个 ArrayList 对象。实际产生的值是一个随机数。产生的随机数被添加到 ArrayList 中。当产生了 10 个数时，通过 complete()方法来结束序列。
     */
    @Test
    public void fun2() {
        Flux.generate(sink -> {
            sink.next("Hello");
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //如果你不调用这个  他会一直执行
            //sink.complete();
        }).subscribe(System.out::println);
    }
    @Test
    public void fun3(){
        final Random random = new Random();
        Flux.generate(ArrayList::new, (list, sink) -> {
            int value = random.nextInt(100);
            list.add(value);
            sink.next(value);
            if (list.size() == 10) {
                sink.complete();
            }else {
                System.out.println("size:"+list.size());
            }
            return list;
        }).subscribe(System.out::println);
    }

    /**
     * create()方法与 generate()方法的不同之处在于所使用的是 FluxSink 对象。FluxSink 支持同步和异步的消息产生，并且可以在一次调用中产生多个元素。
     * 在代码清单 3 中，在一次调用中就产生了全部的 10 个元素。
     */
    @Test
    public void testCreate() throws InterruptedException {
        Flux.create(sink -> {
            for (int i = 0; i < 10; i++) {
                sink.next(i);
            }
            sink.complete();
        }).subscribe(s->{
            System.out.println(s);
        },e ->{
            e.printStackTrace();
        },()->{
            System.out.println("完成");
        });

        Thread.sleep(500);
    }


    @Test
    public void bufferTest(){
        Flux.range(1, 100)
                //buffer 的作用是把 收集成固定 长度的 Flux<List<T>>
                .buffer(20)
                .subscribe(s->{
                    s.size();
                    System.out.println(s);
                });

//        Flux.intervalMillis(100).bufferMillis(1001).take(2).toStream().forEach(System.out::println);
        Flux.range(1, 10)
                // 进入buffer直到 (Predicate.test()) 为止
                .bufferUntil(i -> i % 2 == 0)
                .subscribe(System.out::println);
        Flux.range(1, 10)
                //进入 buffer 当满足(Predicate.test())进去
                .bufferWhile(i -> i % 2 == 0)
                .subscribe(System.out::println);
    }



    @Test
    public void windowTest(){
        Flux.range(1, 100).window(20).subscribe(System.out::println);
    }

    /**
     * zip
     * zipWith 操作符把当前流中的元素与另外一个流中的元素按照一对一的方式进行合并。在合并时可以不做任何处理，由此得到的是一个元素类型为 Tuple2 的流
     * tuple 为元组
     */
    @Test
    public void zipWithTest(){
        Flux.just("a", "b")
                .zipWith(Flux.just("c", "d"))
                .subscribe(System.out::println);
        Flux.just("a", "b")
                .zipWith(Flux.just("c", "d"), (s1, s2) -> String.format("%s-%s", s1, s2))
                .subscribe(System.out::println);

        Flux.just("a", "b")
                .zipWith(Flux.just("c", "d")).subscribe(tuple->{
            String t1 = tuple.getT1();
            String t2 = tuple.getT2();
            System.out.println("t1: " + t1 +", t2 :" +t2);
        });
    }


    /**
     * take 系列操作符用来从当前流中提取元素。提取的方式可以有很多
     * take(long n)，take(Duration timespan)和 takeMillis(long timespan)：按照指定的数量或时间间隔来提取。
     * takeLast(long n)：提取流中的最后 N 个元素。
     * takeUntil(Predicate<? super T> predicate)：提取元素直到 Predicate 返回 true。
     * takeWhile(Predicate<? super T> continuePredicate)： 当 Predicate 返回 true 时才进行提取。
     * takeUntilOther(Publisher<?> other)：提取元素直到另外一个流开始产生元素。
     */
    @Test
    public void takeTest(){
        Flux.range(0,9).takeWhile(i -> i <6).subscribe(System.out::println);
        Flux.range(1, 1000).take(10).subscribe(System.out::println);
        Flux.range(1, 1000).takeLast(10).subscribe(System.out::println);
        Flux.range(1, 1000).takeWhile(i -> i < 10).subscribe(System.out::println);
        Flux.range(1, 1000).takeUntil(i -> i == 10).subscribe(System.out::println);
    }

    /***
     * reduce 和 reduceWith 操作符对流中包含的所有元素进行累积操作，得到一个包含计算结果的 Mono 序列。累积操作是通过一个 BiFunction 来表示的。
     * 在操作时可以指定一个初始值。如果没有初始值，则序列的第一个元素作为初始值。
     * 在代码清单 10 中，第一行语句对流中的元素进行相加操作，结果为 5050；第二行语句同样也是进行相加操作，不过通过一个 Supplier 给出了初始值为 100，所以结果为 5150。
     * 清单 10. reduce 和 reduceWith 操作符使用示例
     */
    @Test
    public void reduceTest(){
        Flux.range(1, 100).reduce((x, y) -> x + y).subscribe(System.out::println);
        // 相对于 上文 就是加上一个 初始值
        Flux.range(1, 100).reduceWith(() -> 100, (x, y) -> x + y).subscribe(System.out::println);
    }



    @Test
    public void pressure() throws InterruptedException {
        Flux<Long> longFlux = Flux.interval(Duration.ofMillis(1));
        longFlux.subscribe(new Subscriber<Long>() {
            Subscription subscription;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Long aLong) {
               // perform(3L);
                subscription.request(10);
                System.out.println("val:"+aLong);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        });
        Thread.sleep(1000);
    }

    @Test
    public void subscribe(){
        Flux<String> stringFlux = Flux.just("Hello","World");
        //stringFlux.subscribe(System.out::println);
        //订阅方式一
        stringFlux.subscribe(val ->{
            System.out.println(val);
        },error ->{
            System.out.println(error);
        },() ->{
            System.out.println("finsh");
        },subscription -> {
            //只订阅一个
            subscription.request(2);
        });
        //订阅方式二
//        stringFlux.subscribe(new Subscriber<String>() {
//            @Override
//            public void onSubscribe(Subscription subscription) {
//                subscription.request(1L);
//            }
//            @Override
//            public void onNext(String s) {
//                System.out.println("onNext:{}"+s);
//            }
//            @Override
//            public void onError(Throwable throwable) {
//            }
//            @Override
//            public void onComplete() {
//                System.out.println("onComplete");
//            }
//        });
    }


    @Test
    public void error() {
        Flux.range(-2, 5)
            .map(val -> {
                int i = val / val;
                System.out.println("map"+i);
                return val;
            })
//            .onErrorContinue((ex, val) -> {  //遇到错误继续订阅
//                if (ex instanceof Exception) {
//                    ex.printStackTrace();
////                    System.out.println( val);
//                } else {
//                }
//
//            })
            //遇到错误，返回新的Flux。继续订阅 和上述 逻辑冲突
            .onErrorResume((ex) -> {
                return Flux.range(10, 16);
            })
            .subscribe(System.out::println);
    }




}
